import com.lecosa.flink.{ChangeSink, MySQLSink}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.datastream.DataStreamSource
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import com.ververica.cdc.connectors.mysql.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import com.ververica.cdc.debezium.DebeziumSourceFunction
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction

import java.sql.{Connection, DriverManager, PreparedStatement}


object MySQLCDC {
  def main(args: Array[String]): Unit = {
    val mySqlSource: DebeziumSourceFunction[String] = MySqlSource.builder[String].hostname("192.168.249.112").port(3306).username("root").password("root").databaseList("shufen").tableList("shufen.student")
      .deserializer(new StringDebeziumDeserializationSchema).startupOptions(StartupOptions.initial).build

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 设置 3s 的 checkpoint 间隔
    env.enableCheckpointing(3000)

    val dataStreamSource: DataStreamSource[String] = env.addSource(mySqlSource)
    dataStreamSource.print


    dataStreamSource.addSink(new ChangeSink)
    env.execute("Print MySQL Snapshot + Binlog")
  }
}
